package cn.com.lyb.flink.source;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class DataGenDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // DataGeneratorSource接口 需要实现，重写map方法，输入类型固定是Long
        DataGeneratorSource<String> stringDataGeneratorSource = new DataGeneratorSource<>(
                new GeneratorFunction<Long, String>() {
                    @Override
                    public String map(Long value) throws Exception {
                        return "Number: " + value;
                    }
                },
                1000000, // 最大值（-1），自动生成的数字序列，从0自增，步长为1
                // RateLimiterStrategy.perSecond(1), // 限速策略每秒生成多少个，限流
                Types.STRING // 返回的类型
        );
        env.fromSource(stringDataGeneratorSource, WatermarkStrategy.noWatermarks(),"dateGen").print();
        env.execute();
    }
}
